package groupOne.app.DWD.db;


import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;


/*
 * 支付成功事务事实表需要将业务数据库中的支付信息表 payment_info 数据与订单明细表关联。
 * 订单明细数据是在下单时生成的，经过一系列的处理进入订单明细主题，通常支付操作在下单后 15min 内完成即可，
 * 因此，支付明细数据可能比订单明细数据滞后 15min。考虑到可能的乱序问题，所以要设置ttl。
 *
 * */
public class DwdTradepayDetail_lyy extends BaseAppSql {
    public static void main(String[] args) {
        new DwdTradepayDetail_lyy().init(4002,2,"DwdTradepayDetail_lyy");
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(30));

        // 1. 下单事务事实表

        tEnv.executeSql("create table dwd_trade_order_detail( " +
                "id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "user_id string, " +
                "split_origin_amount string, " +
                "split_coupon_reduce string, " +
                "split_final_amount string, " +
                "session_id string, " +
                "create_time string, " +
                // "update_time string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "DwdTradepayDetail_lyy"));

        // 2. 支付表 payment_info

        readOdsDb(tEnv, "DwdTradepayDetail_lyy");

        Table paymentInfo = tEnv.sqlQuery("select " +
                        " data['out_trade_no'] out_trade_no, " +
                        " data['order_id'] order_id, " +
                        " data['alipay_trade_no'] alipay_trade_no, " +
                        " data['total_amount'] total_amount, " +
                        " data['trade_body'] trade_body, " +
                        " data['payment_type'] payment_type, " +
                        " data['payment_status'] payment_status, " +
                        " data['create_time'] create_time, " +
                        " data['update_time'] update_time, " +
                        " data['callback_content'] callback_content, " +
                        " data['callback_time'] callback_time, " +
                        " ts " +
                        " from ods_db " +
                        " where `database`='edu' " +
                        " and `table`='payment_info' " +
                        " and (`type`='insert' " +
                        " or  `type`='update') "              //数据模拟只有insert ，目前没法找到updata，这里就不过滤了
                //  "and `old`['payment_status'] is not null " +
                //   "and `data`['payment_status']='1602'
        );

        tEnv.createTemporaryView("payment_info", paymentInfo);


        //2张表join在一起
        Table result = tEnv.sqlQuery("select " +
                "od.id order_detail_id, " +
                "od.course_id, " +
                "od.course_name, " +
                "od.order_id, " +
                "od.user_id, " +
                "od.split_origin_amount, " +
                "od.split_coupon_reduce, " +
                "od.split_final_amount, " +
                "od.session_id, " +
                "od.create_time order_create_time, " +
                // "od.update_time order_update_time, " + //前面orderdetil表没有加这个字段
                "pi.out_trade_no, " +
                "pi.alipay_trade_no, " +
                "pi.total_amount, " +
                "pi.trade_body, " +
                "pi.payment_type, " +
                "pi.payment_status, " +
                "pi.create_time payment_create_time, " +
                "pi.update_time payment_update_time, " +
                "pi.callback_content, " +
                "pi.callback_time, " +
                "pi.ts " +
                "from payment_info pi " +
                "join dwd_trade_order_detail od on pi.order_id=od.order_id ");

        //  tEnv.createTemporaryView("payment_info", result);

        // 5. 写出去
        tEnv.executeSql("create table dwd_trade_pay_detail( " +
                " order_detail_id string, " +
                " course_id string, " +
                " course_name string,  " +
                " order_id string,  " +
                " user_id  string, " +
                " split_origin_amount string,  " +
                " split_coupon_reduce  string, " +
                " split_final_amount string,  " +
                " session_id  string, " +
                " order_create_time string, " +
            //    " order_update_time string,  " +
                " out_trade_no string,  " +
              //  " order_id string,  " +
                " alipay_trade_no string,  " +
                " total_amount string,  " +
                " trade_body string,  " +
                " payment_type string,  " +
                " payment_status string,  " +
                " payment_create_time string,  " +
                " payment_update_time string,  " +
                " callback_content string," +
                " callback_time string,  " +
                " ts bigint " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_DETAIL));

        result.executeInsert("dwd_trade_pay_detail");
    }

    //读取ods_db数据
    public void readOdsDb(StreamTableEnvironment stream,String groupId){
        stream.executeSql("create table ods_db(" +
                " `database` string, " +
                " `table` string, " +
                " `type` string, " +
                " `ts` bigint, " +
                " `data` map<string, string>, " +
                " `old` map<string, string>, " +
                " `pt` as proctime() " +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_ODS_DB, groupId));
    }
}
